package com.roy.minimq.remoting.domain.netty;

import com.roy.minimq.remoting.domain.entity.RemotingCommand;
import com.roy.minimq.remoting.domain.entity.ResponseFuture;
import com.roy.minimq.remoting.domain.entity.common.Pair;
import com.roy.minimq.remoting.domain.util.RemotingHelper;
import com.roy.minimq.remoting.domain.util.RemotingSysResponseCode;
import com.roy.minimq.remoting.exception.RemotingException;
import com.roy.minimq.remoting.service.InvokeCallback;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

public abstract class NettyRemotingAbstract {

    protected Logger logger = LoggerFactory.getLogger(this.getClass());

    protected Pair<NettyRequestProcessor, ExecutorService> defaultRequestProcessor;

    protected final HashMap<Integer/* request code */, Pair<NettyRequestProcessor, ExecutorService>> processorTable =
            new HashMap<Integer, Pair<NettyRequestProcessor, ExecutorService>>(64);
    protected ConcurrentHashMap<Integer, ResponseFuture> responseTable = new ConcurrentHashMap<>();
    protected Semaphore onewaySemaphore;
    protected Semaphore asyncSemaphore;
    protected ExecutorService publicExecutor;

    //TODO RocketMQ中，由于broker即要作为服务端接收请求，又要像客户端一样发送请求并接收响应。所以对请求进行了分类。分为Requst和Response.
    public void processRequest(ChannelHandlerContext ctx, RemotingCommand remotingCommand) {
        RemotingCommand cmd = remotingCommand;
        if(null != cmd){
            if(remotingCommand.isResponseType()){
                processResponseCommand(ctx,cmd);
            }else{
                processRequestCommand(ctx,cmd);
            }
        }
    }
    //和doInvokeAsync方法共同维护responseTable
    protected void processResponseCommand(ChannelHandlerContext ctx, RemotingCommand cmd){
        final int commandId = cmd.getCommandId();
        final ResponseFuture responseFuture = responseTable.get(commandId);
        if(null != responseFuture){
            responseFuture.setResponse(cmd);
            responseTable.remove(commandId);
            if(null != responseFuture.getInvokeCallback()){
                responseFuture.executeCallbackOnce();
            }else{
                responseFuture.putResponse(cmd);
            }
        }else{
            logger.warn("receive response, but not matched any request, " + RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
            logger.warn(cmd.toString());
        }
    }

    protected void processRequestCommand(ChannelHandlerContext ctx, RemotingCommand cmd){
        final Pair<NettyRequestProcessor, ExecutorService> matched = processorTable.get(cmd.getCode());
        final Pair<NettyRequestProcessor, ExecutorService> executePair = null == matched?this.defaultRequestProcessor:matched;

        if(null != executePair){
            if(executePair.getObject1().rejectRequest()){
                final RemotingCommand responseCommand = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY
                        , "Request is rejected by server.wait for a while");
                ctx.writeAndFlush(responseCommand);
                return;
            }
            final ExecutorService executorService = executePair.getObject2();
            final NettyRequestProcessor processor = executePair.getObject1();
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        final RemotingCommand responseCommand = processor.processRequest(ctx, cmd);
                        if(!cmd.isOnewayRPC() && null != responseCommand){
                            responseCommand.markResponseType();
                            ctx.writeAndFlush(responseCommand);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
        }else{
            String error = " request type " + cmd.getCode() + " not supported";
            final RemotingCommand response =
                    RemotingCommand.createResponseCommand(RemotingSysResponseCode.REQUEST_CODE_NOT_SUPPORTED, error);
            ctx.writeAndFlush(response);
            logger.error("request type error from "+ctx.channel().remoteAddress());
        }
    }

    protected RemotingCommand doInvokeSync(Channel channel, RemotingCommand request, long timeoutMillis) throws InterruptedException, RemotingException {
        ResponseFuture responseFuture = new ResponseFuture(channel, timeoutMillis, null);
        this.responseTable.put(request.getCommandId(), responseFuture);
        channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                    responseFuture.setSendMessageOk(true);
                    return;
                } else {
                    responseFuture.setSendMessageOk(false);
                    logger.warn("command[id={}] send to {} failed", request.getCommandId(), channel.remoteAddress());
                    responseFuture.putResponse(null);
                    return;
                }

            }
        });
//        responseFuture.putResponse(request);
        RemotingCommand responseCommand = responseFuture.waitResponse();
        if (null == responseCommand) {
            throw new RemotingException("command[id=" + request.getCommandId() + "] send to " + channel.remoteAddress() + " failed");
        }
        return responseCommand;
    }

    protected void doInvokeOneway(Channel channel, RemotingCommand request, long timeoutMillis) throws InterruptedException, RemotingException {
        //使用semaphore控制下频率
        boolean acquire = this.onewaySemaphore.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
        if (acquire) {
            try {
                channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        onewaySemaphore.release();
                        if (!future.isSuccess()) {
                            logger.warn("send a request command to channel <" + channel.remoteAddress() + "> failed.");
                        }
                    }
                });
            } catch (Exception e) {
                this.onewaySemaphore.release();
                throw new RemotingException("send request command to channel failed", e);
            }
        } else {
            String info = String.format(
                    "doInvokeOneway tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d",
                    timeoutMillis,
                    this.onewaySemaphore.getQueueLength(),
                    this.onewaySemaphore.availablePermits()
            );
            logger.warn(info);
            throw new RemotingException(info);
        }
    }

    //正常的业务回调会在Handler中注入的processor中处理。这里传的invokeCallback相当于是失败后的兜底方法。
    protected void doInvokeAsync(Channel channel, RemotingCommand request, long timeoutMillis, InvokeCallback invokeCallback) throws InterruptedException, RemotingException {
        //使用semaphore控制下频率
        boolean acquire = this.asyncSemaphore.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
        if (acquire) {
            try {
                ResponseFuture responseFuture = new ResponseFuture(channel, timeoutMillis, invokeCallback);
                int commandId = request.getCommandId();
                this.responseTable.put(commandId, responseFuture);
                channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        asyncSemaphore.release();
                        // 正常的业务流程在上面的processResponsCommand中处理
                        // 这里只处理失败的请求。
                        if (future.isSuccess()) {
                            responseFuture.setSendMessageOk(true);
                            return;
                        }
                        ResponseFuture response = responseTable.remove(commandId);
                        if (response != null) {
                            response.setSendMessageOk(false);
                            response.putResponse(null);
                            try {
                                executeInvokeCallback(response);
                            } catch (Throwable e) {
                                logger.warn("execute callback in requestFail, and callback throw", e);
                            }
                        }
                    }
                });
            } catch (Exception e) {
                this.asyncSemaphore.release();
                logger.warn("send a request command to channel {} Exception", channel.remoteAddress(), e);
                throw new RemotingException("send request command to channel failed", e);
            }
        } else {
            String info = String.format(
                    "doInvokeAsync tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d",
                    timeoutMillis,
                    this.asyncSemaphore.getQueueLength(),
                    this.asyncSemaphore.availablePermits()
            );
            logger.warn(info);
            throw new RemotingException(info);
        }
    }
    //执行一次回调函数
    private void executeInvokeCallback(ResponseFuture responseFuture) {
        this.publicExecutor.submit(() -> {
            responseFuture.executeCallbackOnce();
        });
    }
}
